package customerproportion.sparkSql;

import customerproportion.User.UserInfo;
import customerproportion.User.UserLog;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class GetCustomerProportionSparkSql {

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("Get CustomerProportionSparkSql").getOrCreate();
        String UserInfoFileName = "./src/main/resources/data_format1/data_format1/user_info_format1.csv";
        String UserLogFileName = "./src/main/resources/data_format1/data_format1/user_log_format1.csv";
/*
        StructType userInfoSchema = new StructType(new StructField[]{
                new StructField("user_id", DataTypes.StringType, false, Metadata.empty()),
                new StructField("age_range", DataTypes.StringType, true, Metadata.empty()),
                new StructField("gender", DataTypes.StringType, true, Metadata.empty())
        });

        Dataset<Row> TempUserInfo = spark.read()
                .schema(userInfoSchema)
                .csv(UserInfoFileName)
                .map((MapFunction<Row, Row>) row -> {
                    String user_id = row.getString(0);
                    String age_range = row.getString(1);
                    String gender = row.getString(2);
                    if (age_range.equals("")) age_range = "0";
                    if (gender.equals("")) gender = "2";
                    return RowFactory.create(user_id, age_range, gender);
                }, Encoders.javaSerialization(Row.class));
        TempUserInfo.printSchema();
        Dataset<Row> UserInfo = spark.createDataFrame(TempUserInfo.rdd(), userInfoSchema);
 */
        Dataset<UserInfo> UserInfo = spark.read().textFile(UserInfoFileName)
                .map((MapFunction<String, UserInfo>) UserInfo::new, Encoders.bean(UserInfo.class));
        UserInfo.printSchema();
        UserInfo.createOrReplaceTempView("UserInfo");
/*
        StructType userLogSchema = new StructType(new StructField[]{
                new StructField("user_id", DataTypes.StringType, false, Metadata.empty()),
                new StructField("item_id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("cat_id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("seller_id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("brand_id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("time_stamp_id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("action_type", DataTypes.IntegerType, false, Metadata.empty())
        });
*/
        Dataset<UserLog> UserLog = spark.read().textFile(UserLogFileName)
                .map((MapFunction<String, UserLog>) UserLog::new, Encoders.bean(UserLog.class));

        UserLog.printSchema();
        UserLog.createOrReplaceTempView("UserLog");

        Dataset<Row> CustomerId = spark.sql(
                "SELECT DISTINCT log.UserId FROM UserLog log " +
                        "WHERE log.ActionType=2");
        CustomerId.printSchema();
        CustomerId.createOrReplaceTempView("CustomerId");


        Dataset<Row> CustomerInfo = spark.sql(
                "SELECT * FROM UserInfo info " +
                        "WHERE info.UserId IN " +
                        "(SELECT * FROM CustomerId)");
        CustomerInfo.printSchema();
        CustomerInfo.createOrReplaceTempView("CustomerInfo");

        System.out.println(spark.sql("SELECT AgeRange,COUNT(*) AS count " +
                "FROM CustomerInfo " +
                "WHERE AgeRange!=0 " +
                "GROUP BY AgeRange " +
                "SORT BY AgeRange").collectAsList().toString());
        System.out.println(spark.sql("SELECT Gender,COUNT(*) AS count " +
                "FROM CustomerInfo " +
                "WHERE Gender!=2 " +
                "GROUP BY Gender " +
                "SORT BY Gender").collectAsList().toString());
//        CustomerInfo.groupBy("age_range").count().show();
//        CustomerInfo.groupBy("gender").count().show();
    }
}
